package hit.edu.zjc.Operator;


import hit.edu.zjc.Tool.ObjectConvert;
import hit.edu.zjc.Tool.RWObject;
import hit.edu.zjc.Tool.ShowOutPut;
import hit.edu.zjc.marshalling.ByteSerializerDeserializer;
import hit.edu.zjc.parsers.ByteTupleParserFactory;

import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.IOperatorDescriptorRegistry;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;

/*
 * 将PartitionSetList，根据SetTable，找到真正的重复记录
 * 是实体识别的最后一步。
 * */
public class SourceMatchOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
    private static final long serialVersionUID = 1L;
    private String SetListFilePath;//含有ArrayList<HashSet<Integer>>的二进制的文件路径
    private String SetTableFilePath;//含有Hashtable<Integer,HashSet<Integer>>的二进制文件路径
    private double Threshold;
    public SourceMatchOperatorDescriptor(IOperatorDescriptorRegistry spec,String setlistfilepath,String settablefilepath,double th) {
        super(spec, 0, 1);  
        recordDescriptors[0]  = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});
        SetListFilePath=setlistfilepath;
        SetTableFilePath=settablefilepath;
        Threshold=th;
    }

    @Override
    public IOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
    	 final ITupleParser parser =new ByteTupleParserFactory().createTupleParser(ctx);
    	return new AbstractUnaryOutputSourceOperatorNodePushable() {
        	
    		private ArrayList<HashSet<Integer>> SetList;
        	private Hashtable<Integer,HashSet<Integer>> SetTable;
        	ArrayList<HashSet<Integer>> EntityList=new ArrayList<HashSet<Integer>>();//最终的实体列表。
        	    	
        	  //获得两个集合交集的元素的数量
          public int GetCommonNumber(HashSet<Integer>set1,HashSet<Integer>set2)
               {
            	   Iterator<Integer> iterator=set2.iterator();
            	   int Sum=0;
            	   while(iterator.hasNext())
            	   {
            		   Integer id=iterator.next();
            		   if(set1.contains(id))
            			   Sum++;
            	   }
            	   return Sum;
               }
            //获得两个集合并集的数量。
          public int GetTotalNumber(HashSet<Integer>set1,HashSet<Integer>set2)
               {
            		int cover=GetCommonNumber(set1,set2);
            		return (set1.size()+set2.size()-cover);
               }
       
          public int SearchSIndex(Integer value,ArrayList<HashSet<Integer>> S)
          	{
        	  	for(int index=0;index<S.size();index++)
        	  	{
        	  		HashSet<Integer>set=S.get(index);
        	  		if(set.contains(value))
        	  			return index;
        	  	}
        	  	return -1;
        	  
          	}
              
            @Override
            public void initialize() throws HyracksDataException {
            	
            	//读取分配到该NC上的SetList
            	SetList=(ArrayList<HashSet<Integer>>)RWObject.ReadObject(SetListFilePath);
            	//读取所有的连接信息SetTable
            	SetTable=(Hashtable<Integer,HashSet<Integer>>)RWObject.ReadObject(SetTableFilePath);   
            	
            	ArrayList<HashSet<Integer>> N;//节点的邻居节点。每个节点一个
            	ArrayList<HashSet<Integer>> S;//节点合并的结合。
            	for(int i=0;i<SetList.size();i++)
            	{
            		HashSet<Integer> set=SetList.get(i);
            		N=new ArrayList<HashSet<Integer>>();
            		S=new ArrayList<HashSet<Integer>>();
            		//开始初始化N，S
            		Iterator<Integer> iterator=set.iterator();
            		while(iterator.hasNext())
            		{
            			Integer id=iterator.next();
            			HashSet<Integer> initset=new HashSet<Integer>();
            			initset.add(id);
            			S.add(initset);
            			HashSet<Integer> eageset=SetTable.get(id);
            			if(eageset==null)
            			{
            				eageset=new HashSet<Integer>();
            				eageset.add(id);
            			}
            			N.add(eageset);
            		} 		
            		//更新其他N,现在S中的每个set只有一个元素
            		for(int k=0;k<S.size();k++)
            		{
            			HashSet<Integer> idset=S.get(k);
            			Iterator<Integer> it=idset.iterator();
            			Integer id=it.next();
            			HashSet<Integer> eageset=N.get(k);
            			it=eageset.iterator();
            			ArrayList<Integer> DeleteList=new ArrayList<Integer>();
            			while(it.hasNext())
            			{
            				Integer otherid=it.next();
            				if(otherid.intValue()==id.intValue())
            					continue;
            				int index=SearchSIndex(otherid,S);
            				HashSet<Integer>othereageset=N.get(index);
            				//othereageset.add(id);//两个节点，只要一方达到Threshold，就构成了边。
            				boolean isContain=othereageset.contains(id);
            				if(!isContain)
            				{
            					//eageset.remove(otherid);//必须两个节点相互连接，才能构成边。
            					DeleteList.add(otherid);
            				}
            			}
            			for(int m=0;m<DeleteList.size();m++)  //去掉for循环，就构成了有向图
            				eageset.remove(DeleteList.get(m));
            		}
            		
            		//开始实体识别
            		boolean flag=true;
            		while(flag)
            		{
            			flag=false;
            			for(int u=S.size()-1;u>=0;u--)
            			{
            				for(int v=u-1;v>=0;v--)
            				{
                			HashSet<Integer> u_N=N.get(u);
            					HashSet<Integer> v_N=N.get(v);	
            					HashSet<Integer> u_S=S.get(u);
            					HashSet<Integer> v_S=S.get(v);
            					
            					//检验第一种情况。
            					boolean isContain=v_S.containsAll(u_N)||u_S.containsAll(v_N);
            					
            					//检验第二种情况。
            					if(!isContain)
            					{
            						int covernum=GetCommonNumber(u_N,v_N);
                				int totalnum=GetTotalNumber(u_N,v_N);
                				double rate=1.0*covernum/totalnum;
                				if(rate<Threshold)
                					continue;
            					}
            					
            					//执行合并操作
            					flag=true;
            					v_N.addAll(u_N);
            					v_S.addAll(u_S);
            					S.remove(u);
            					N.remove(u);
            					break;
            				}				           				
            			}	
            		}
            		EntityList.addAll(S); 
            	}
            	for(int index=EntityList.size()-1;index>=0;index--)
            	{
            		if(EntityList.get(index).size()<=1)
            			EntityList.remove(index);
            	}
            	
            	
            	//完成所有重复记录检测工作，开始写出。
             writer.open();
             try
             	{
	     	    	byte[] SetListBytes =ObjectConvert.ObjectToByteArray(EntityList);  		   
	     		   ByteArrayInputStream bais = new ByteArrayInputStream(SetListBytes);	
	     	    	parser.parse(bais, writer);
	     	      bais.close();
	            
             	}
             catch(Exception e)
             	{
            	 	writer.fail();
            	 	e.printStackTrace();
             	}
             finally
             	{
            	 	writer.close();
             	}
             System.gc();
            }
        };
    }
}